package com.gxkj.spark;
import org.apache.spark.api.java.function.CoGroupFunction;
import scala.Tuple2;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.SparkSession;

import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;
import org.apache.spark.sql.SparkSession;

/**
 * @author sqyc
 * 入门：http://spark.apache.org/docs/latest/rdd-programming-guide.html
 *
 * 这个例子的目标是统计词语出现的频次
 */
public final class JavaWordCount {
    private static final Pattern SPACE = Pattern.compile(" ");

    public static void main(String[] args) throws Exception {


        args = new String[]{"D:\\wp\\open_movie\\server\\spark_demo\\src\\main\\resources\\demo.txt"};

        if (args.length < 1) {
            System.err.println("Usage: JavaWordCount <file>");
            System.exit(1);
        }

        /**
         * 初始化一个会话
         */
        SparkSession spark = SparkSession
                .builder()
                .appName("JavaWordCount")
                .getOrCreate();

        /**
         * textFile方法返回url里的文本的行的集合。
         */
        JavaRDD<String> lines = spark.read().textFile(args[0]).javaRDD();

        /**
         * 统计长度
         * map:是一个转换函数，将dataset的元素传递到一个函数里，并返回一个新的RDD集合。
         * reduce ：是一个action行动函数，将RDD里所有的元素使用一个函数聚合在一起，返回一个最终结果。
         *
         * RDDs支持两种类型的操作，一种是转换（如map方法）,一种是动作（actions，如reduce）
         */
      Integer lengthSum =  lines.map(s -> s.length()).reduce((a, b) -> a + b);
      System.out.println("lengthSum="+lengthSum);

        /**
         * 将每行出现的词放到JavaRDD里
         *
         * flatMap方法实现一对多，显示一个字符流。参考https://blog.csdn.net/liyantianmin/article/details/96178586
         */
        JavaRDD<String> words = lines.flatMap(
                s -> Arrays.asList(SPACE.split(s)).iterator()
        );

        /**
         * 键值对操作
         */
        JavaPairRDD<String, Integer> ones = words.mapToPair(
                s -> new Tuple2<>(s, 1)
        );

        /**
         *
         * 按照键值对进行聚合
         */
        JavaPairRDD<String, Integer> counts = ones.reduceByKey((i1, i2) -> i1 + i2);

        /**
         * 将结果集进行汇总
         * To collect the word counts in our shell, we can call collect:
         */
        List<Tuple2<String, Integer>> output = counts.collect();
        /**
         * 打印输出
         */
        for (Tuple2<?,?> tuple : output) {
            System.out.println(tuple._1() + ": " + tuple._2());
        }
        spark.stop();
    }
}
